package com.roc.rocketmqconsumer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RequestReplyMsgConsumer {
    private static final String REQUEST_TOPIC = "TOPIC_REQUEST";
    private static final String REPLY_TOPIC = "TOPIC_REPLY";
    private static final String GROUP_ID = "roc_consumer_group";

    public static void main(String[] args) throws MQClientException {
        // 初始化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_ID);
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe(REQUEST_TOPIC, "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String requestId = msg.getUserProperty("requestId");
                        String requestMessage = new String(msg.getBody(), "UTF-8");
                        System.out.println("收到请求消息：" + requestMessage);

                        // 模拟处理请求消息
                        String replyMessage = "处理结果：请求消息处理成功";

                        // 发送应答消息
                        DefaultMQProducer replyProducer = new DefaultMQProducer(GROUP_ID + "_reply_producer");
                        replyProducer.setNamesrvAddr("localhost:9876");
                        replyProducer.start();

                        Message reply = new Message(REPLY_TOPIC, replyMessage.getBytes());
                        reply.putUserProperty("requestId", requestId);
                        replyProducer.send(reply);
                        replyProducer.shutdown();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("消费者已启动，等待接收请求消息");
    }
}